package com.application.service.Impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.BooleanUtil;
import com.application.controller.Result;
import com.application.dao.ActivityMemberDao;
import com.application.domain.ActivityMember;
import com.application.dto.UserDTO;
import com.application.service.ActivityMemberService;
import com.application.service.ActivityService;
import com.application.util.RedisIdWorker;
import com.application.util.UserHolder;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import sun.security.krb5.internal.APOptions;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.applet.AppletContext;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static com.application.util.RedisConstant.LOCK_USER_KEY;

@Service
public class ActivityMemberServiceImpl extends ServiceImpl<ActivityMemberDao, ActivityMember> implements ActivityMemberService {

    @Resource
    private RedisIdWorker redisIdWorker;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Resource
    private ActivityService activityService;

    @Resource
    private RedissonClient redissonClient;

    /**
     * 这里就是  预编译lua脚本  因为 我们判断人数和减免人数操作  需要保持原子性
     */
    private final static DefaultRedisScript<Long> APPLY_ACTIVITY;
    static {
        APPLY_ACTIVITY = new DefaultRedisScript<>();
        // 首先 找到 lua脚本进行 预加载
        APPLY_ACTIVITY.setLocation(new ClassPathResource("applyActivity.lua"));
        // 确定返回值的类型
        APPLY_ACTIVITY.setResultType(Long.class);
    }


    // 这里因为 我们进入数据更新时候  涉及了 事务操作
    // 但是 spring只会  管理 主线程  我们是使用 子线程来异步更新数据库的  所以 需要在父类这里赋值受管理的代理对象
    private ActivityMemberService proxy;

    // 这里是 controller 进来的接口
    // 这里实现  用户 进行抢 活动
     @Override
     /**
      * 这里实现的功能 就是 接收前端传来的 需要参加的  用户和活动信息
      * 然后 获取对应的唯一id  并且启动子线程来 异步 更新redis和数据库
      */
    public Result applyActivity(String activityId) {
         // 获取对应的用户Id   这里 就是 获取当前用户的id  因为 我们分开测试 所以先写死
//         String userId = UserHolder.getUser().getId();

         String userId = "1";
         // 首先生成 这个 加入 活动 的  唯一id
         long nextId = redisIdWorker.nextId("activity");

         // 然后使用lua脚本来判断是否符合参加条件
         // 要么没有满人   要么 没有重复参加
         // 这里就是  执行lua脚本
         Long result = stringRedisTemplate.execute(
                 APPLY_ACTIVITY,
                 // 这里就是 lua脚本中的 KEYS[] 数组
                 Collections.emptyList(),
                 // 这里就是 lua脚本的 AVGS[]
                 // 传进去  用户id  活动id  以及 生成记录对应的随机id
                 userId, activityId,
                 String.valueOf(nextId)
         );

         if(result.intValue() != 0){
             String msg = result == 2 ? "请不要重复参加" :"人数已满";
             Integer code = 500;
             return new Result(code,result,msg);
         }

         // 这里就是 获取代理对象  给 子线程使用
         proxy = (ActivityMemberService) AopContext.currentProxy();
         return new Result(200,nextId,"参加活动成功");
    }


    /**
     * 下面就是  用来异步接收  redis当中的消息 以及 放入数据库
     */
    // 创建一个 单线程 用来 处理
    private static final ExecutorService APPLY_ACTIVITY_EXECUTOR = Executors.newSingleThreadExecutor();
    // 这个注解 就是当前类 初始化之后  就开始执行这个方法
    @PostConstruct
    private void init(){
        // 然后 我们使用 内部类 来实现代码
        // 其实 代码 就似乎 不断循环到  redis当中的  队列中  查看是否有消息
        APPLY_ACTIVITY_EXECUTOR.submit(new ApplyActivity());
    }

    /**
     * 这里就是  创建内部类  用来实现  异步处理功能
     */
    private class ApplyActivity implements Runnable{
        @Override
        public void run() {
            while (true){
                try {
                    // 首先 到消息队列当中 获取信息
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            // 创建消费者组  而且 指定消费者
                            Consumer.from("g1", "c1"),
                            // 这里就是 读取信息时候的可选择
                            //  如果null值  不要抛出异常 返回空    读取一个消息     阻塞等待2秒
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create("stream.activity", ReadOffset.lastConsumed())
                    );

                    if(list == null || list.isEmpty()){
                        // 如果没有消息  那么继续循环等待消息
                        continue;
                    }

                    /**
                     * 有消息  就取出消息里面的详细信息   进行处理
                     */
                    // 因为我们只是取一个消息  所以 直接取 下标为0
                    MapRecord<String, Object, Object> record = list.get(0);
                    // 获取里面的数据
                    Map<Object, Object> value = record.getValue();
                    // 然后进行转换为 对应的实体类
                    ActivityMember activityMember = BeanUtil.fillBeanWithMap(value, new ActivityMember(), true);
                    System.err.println(activityMember);
                    // 有消息 就说明 redis中 人数减1  用户加入了 集合
                    // 所以 可以  更新数据库了  (异步)
                    handleApplyActivity(activityMember);

                    // 处理完之后  需要 返回 ack  确认消息消费完  record.getId 就是 我们使用 * 生成随机的消息id
                    stringRedisTemplate.opsForStream().acknowledge("stream.activity","g1",record.getId());
                }catch (Exception e) {
                    log.error("消息处理功能异常",e);
                    handlePendinglist(); // 这里就是异常情况的时候  可能  消息队列中消息还没有ack  所以  这里就是ack
                }
            }
        }

        private void handlePendinglist(){
            while (true){
                try {
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create("stream.activity", ReadOffset.from("0"))
                    );
                    // 没有消息 那么 就不需要 我们 取确认了
                    if(list == null || list.isEmpty()){
                        break;
                    }

                    /**
                     * 这里就是  处理 list中的数据
                     */
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> value = record.getValue();
                    ActivityMember activityMember = BeanUtil.fillBeanWithMap(value, new ActivityMember(), true);
                    // 下订单
                    handleApplyActivity(activityMember);

                    // ack
                    stringRedisTemplate.opsForStream().acknowledge("stream.activity","g1",record.getId());
                }catch (Exception e){
                    log.error("消息消费异常",e);
                    // 如果发生异常  那么 就 sleep20毫秒 然后 继续 到  pendinglist中继续处理
                    try {
                        Thread.sleep(20L);
                    } catch (InterruptedException ex) {
                        throw new RuntimeException(ex);
                    }
                }
            }
        }
    }


    /**
     * 这里才是 正在  到数据库操作
     * @param activityMember
     */
    public void handleApplyActivity(ActivityMember activityMember){
        // 这里我们还是需要 使用分布式锁来 保证同一时间 同一个用户  参加活动
        //TODO 先使用  redis中  自带那个  到时  再用 redisson
        // 这里我们可以使用 redisson

        RLock lock = redissonClient.getLock("activity:member:" + activityMember.getUserId());

        if(!lock.tryLock()){
            log.error("请不要重复下单");
            return;
        }
        try {
            // 这里就是 使用 spring管理主线程的 代理对象 进行 调用方法   实现事务管理
            // 不然 因为这里是子线程执行   spring无法管理
            proxy.createApplyActivity(activityMember);
        }finally {
            lock.unlock();
        }
    }

    /**
     * 这里就是 实现到数据库实现
     * @param activityMember
     */
    @Transactional
    public void createApplyActivity(ActivityMember activityMember){
        String userId = activityMember.getUserId();
        String activityId = activityMember.getActivityId();
        // 为了更加稳  可以到 数据库查询 是否 有相等数据
        int count = query().eq("user_id", userId).eq("activity_id", activityId).count();

        if(count > 0){
            log.error("请不要重复下单");
            return;
        }

        // 首先就是 到 活动表中对应 活动的人数 进行减少  然后 在  本表中进行生成记录
        boolean isSuccess = activityService.update().setSql("limit_number = limit_number - 1")
                .eq("ac_id", activityMember.getActivityId())
                // 这里就是  nums > 0
                .gt("limit_number", 0)
                .update();

        if(!isSuccess){
            log.error("库存不足");
            return;
        }
        save(activityMember);
    }

    public boolean tryLock(String key){
        Boolean isSuccess = stringRedisTemplate.opsForValue().setIfAbsent(LOCK_USER_KEY + key, "1");
        return BooleanUtil.isTrue(isSuccess);
    }

    public void unLock(String key){
        stringRedisTemplate.delete(LOCK_USER_KEY + key);
    }
}
